package cn.jly.flink.state;

import cn.jly.flink.entity.SensorReading;
import cn.jly.flink.utils.FlinkUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * 利用keyedState，实现需求：监测传感器的温度值，如果传感器连续两个温度差值超过10度，就给出报警
 *
 * @PackageName cn.jly.flink.state
 * @ClassName PracticeDemo
 * @Description 状态编程练习
 * @Author 姬岚洋
 * @Date 2021/2/1 下午 7:16
 */
public class PracticeDemo {
    private static final String HIGH_TEMP_DIFF_TAG = "high_temp_diff";

    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = FlinkUtils.getStreamExecutionEnv();

        final ParameterTool parameterTool = ParameterTool.fromArgs(args);
        final String host = parameterTool.get("host", "192.168.172.101");
        final int port = parameterTool.getInt("port", 9999);

        // 温差报警通过侧输出流输出
        final OutputTag<Tuple3<String, SensorReading, SensorReading>> highTempDiffTag = new OutputTag<>(HIGH_TEMP_DIFF_TAG, TypeInformation.of(new TypeHint<Tuple3<String, SensorReading, SensorReading>>() {
        }));

        final SingleOutputStreamOperator<Tuple2<String, SensorReading>> processDataStream = env.socketTextStream(host, port)
                .flatMap(new FlatMapFunction<String, SensorReading>() {
                    @Override
                    public void flatMap(String value, Collector<SensorReading> out) throws Exception {
                        if (StringUtils.isNotEmpty(value)) {
                            final String[] split = value.split(",");
                            out.collect(new SensorReading(split[0], Long.parseLong(split[1]), Double.valueOf(split[2])));
                        }
                    }
                })
                .keyBy(new KeySelector<SensorReading, String>() {
                    @Override
                    public String getKey(SensorReading value) throws Exception {
                        return value.getId();
                    }
                })
                .process(new KeyedProcessFunction<String, SensorReading, Tuple2<String, SensorReading>>() {
                    // 状态，存储上一次温度值
                    transient ValueState<SensorReading> valueState = null;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);

                        // 初始化状态
                        final ValueStateDescriptor<SensorReading> valueStateDescriptor = new ValueStateDescriptor<>(
                                "温度状态",
                                TypeInformation.of(new TypeHint<SensorReading>() {
                                })
                        );
                        valueState = getRuntimeContext().getState(valueStateDescriptor);
                    }

                    @Override
                    public void processElement(SensorReading value, Context ctx, Collector<Tuple2<String, SensorReading>> out) throws Exception {
                        // 取出状态值
                        final SensorReading lastSensorReading = valueState.value();

                        if (lastSensorReading != null) {
                            final double tempDiff = Math.abs(value.getTemperature() - lastSensorReading.getTemperature());
                            if (tempDiff > 10d) {
                                ctx.output(highTempDiffTag, Tuple3.of("两次温差超过10度", lastSensorReading, value));
                            }
                        }

                        // 更新状态
                        valueState.update(value);

                        // 输出正常温度
                        out.collect(Tuple2.of("当前温度", value));
                    }
                });

        // 正常输出
        processDataStream.print("正常");

        // 获取警报信息
        processDataStream.getSideOutput(highTempDiffTag).printToErr("警报");

        FlinkUtils.executeStream(env, "PracticeDemo");
    }
}
